package com.shujia.spark.opt

import java.sql.{Connection, DriverManager, PreparedStatement}

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo1ForeachPartition {
  def main(args: Array[String]): Unit = {

    val conf: SparkConf = new SparkConf()
      .setMaster("local")
      .setAppName("partition")

    val sc = new SparkContext(conf)


    val studentRDD: RDD[String] = sc.textFile("data/students.txt")

    /**
      * 1、将创建链接放在Driver端，在Executor不能使用这个链接，网络链接不能在网络中传输
      * 2、将创建链接放在foreach里面，每一条数据都会创建一个链接，需要创建很多链接，效率低，而且会导致mysql链接不够
      *
      */

    /**
      * 将rdd的数据保存到mysql中
      *
      */
    /*

        studentRDD.foreach(line => {
          //写jdbc代码
          Class.forName("com.mysql.jdbc.Driver")
          //1、创建链接
          val con: Connection = DriverManager.getConnection("jdbc:mysql://master:3306/bigdata?useUnicode=true&characterEncoding=utf-8", "root", "123456")

          //编写sql
          val stat: PreparedStatement = con.prepareStatement("insert into students(id,name,age,gender,clazz) values(?,?,?,?,?)")

          val split: Array[String] = line.split(",")

          //设置列值
          stat.setString(1, split(0))
          stat.setString(2, split(1))
          stat.setInt(3, split(2).toInt)
          stat.setString(4, split(3))
          stat.setString(5, split(4))


          //执行插入
          stat.execute()

          con.close()

        })
    */

    /**
      * foreachPartition: 一次还进去一个分区的数据
      * 为每一个分区创建一个链接
      *
      *
      * 当我们将rdd的数据写入到外部数据库中的时候需要使用foreachPartition
      *
      */


    studentRDD.foreachPartition(iter => {

      println("创建链接")
      Class.forName("com.mysql.jdbc.Driver")
      //1、创建链接
      val con: Connection = DriverManager.getConnection("jdbc:mysql://master:3306/bigdata?useUnicode=true&characterEncoding=utf-8", "root", "123456")

      iter.foreach(line => {
        //编写sql
        val stat: PreparedStatement = con.prepareStatement("insert into students(id,name,age,gender,clazz) values(?,?,?,?,?)")

        val split: Array[String] = line.split(",")

        //设置列值
        stat.setString(1, split(0))
        stat.setString(2, split(1))
        stat.setInt(3, split(2).toInt)
        stat.setString(4, split(3))
        stat.setString(5, split(4))


        //执行插入
        stat.execute()
      })

    })


    while (true) {

    }
  }

}
